﻿#include "pch.h"

using namespace std::chrono_literals;
namespace fs = boost::filesystem;

#ifdef _WIN32
#pragma comment(lib, "mysqlcppconn.lib")
#endif

std::string to_upper(const std::string &input)
{
    std::string result = input;
    std::transform(result.begin(), result.end(), result.begin(), ::toupper);
    return result;
}

std::string to_lower(const std::string &input)
{
    std::string result = input;
    std::transform(result.begin(), result.end(), result.begin(), ::tolower);
    return result;
}

// 日志级别转字符串
std::string log_level_string(LogLevel level)
{
    static const std::array<const char *, 6> level_names = {"TRACE", "DEBUG", "INFO", "WARNING", "ERROR", "FATAL"};
    return level_names[static_cast<int32_t>(level)];
}

// 线程安全的本地时间转换函数（POSIX 系统）
std::tm localtime_thread_safe(const std::time_t &tt)
{
    std::tm result;
#ifdef _POSIX_C_SOURCE
    // POSIX 系统使用 localtime_r
    localtime_r(&tt, &result);
#else
    // 非 POSIX 系统（如 Windows）使用互斥锁保护 localtime
    static std::mutex mtx;
    std::lock_guard<std::mutex> lock(mtx);
    result = *std::localtime(&tt);
#endif
    return result;
}

LogLevel parse_log_level(const std::string &level_str)
{
    std::string upper_str = to_upper(level_str);
    if (upper_str == "TRACE")
        return LogLevel::TRACE_;
    if (upper_str == "DEBUG")
        return LogLevel::DEBUG_;
    if (upper_str == "WARNING")
        return LogLevel::WARNING_;
    if (upper_str == "ERROR")
        return LogLevel::ERROR_;
    if (upper_str == "FATAL")
        return LogLevel::FATAL_;
    return LogLevel::INFO_;
}

namespace
{
    LogLevel string_to_level(const std::string &level)
    {
        static const std::unordered_map<std::string, LogLevel> level_map = {{"TRACE", LogLevel::TRACE_}, {"DEBUG", LogLevel::DEBUG_}, {"INFO", LogLevel::INFO_}, {"WARNING", LogLevel::WARNING_}, {"ERROR", LogLevel::ERROR_}, {"FATAL", LogLevel::FATAL_}};
        auto it = level_map.find(to_upper(level));
        return it != level_map.end() ? it->second : LogLevel::INFO_;
    }

    // std::string timestamp_to_string(const std::chrono::system_clock::time_point &tp)
    // {
    //     std::time_t tt = std::chrono::system_clock::to_time_t(tp);
    //     std::tm local_tm = localtime_thread_safe(tt);
    //     return (boost::format("%04d%02d%02d_%02d%02d%02d") % (local_tm.tm_year + 1900) % (local_tm.tm_mon + 1) % local_tm.tm_mday % local_tm.tm_hour % local_tm.tm_min % local_tm.tm_sec).str();
    // }
} // namespace

// ConfigManager实现
ConfigManager::ConfigManager(sw::redis::Redis &redis) : redis_(redis) {}

void ConfigManager::refresh()
{
    std::cout << "ConfigManager::refresh." << std::endl;
    std::lock_guard<std::mutex> lock(mutex_);
    try
    {
        std::vector<std::string> keys;
        redis_.keys("logging:config:*", std::back_inserter(keys));
        for (const auto &key : keys)
        {
            const std::string channel = std::string("logging:channel:") + key.substr(15);

            std::unordered_map<std::string, std::string> fields;
            redis_.hgetall(key, std::inserter(fields, fields.begin()));

            const auto new_config = parse_config(fields);

            std::cout << "ConfigManager::refresh. channel=" << channel << ", config = " << new_config << std::endl;
            if (const auto it = configs_.find(channel); it == configs_.end() || it->second != new_config)
            {
                configs_[channel] = new_config;
                pending_changes_.insert(channel);
            }
        }
    }
    catch (const sw::redis::TimeoutError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis TimeoutError error: " << e.what() << std::endl;
    }
    catch (const sw::redis::IoError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis IoError error: " << e.what() << std::endl;
    }
    catch (const std::exception &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "General error: " << e.what() << std::endl;
    }
}

OutputConfig ConfigManager::get_config(const std::string &channel) const
{
    std::lock_guard<std::mutex> lock(mutex_);
    return configs_.count(channel) ? configs_.at(channel) : OutputConfig{};
}

std::unordered_set<std::string> ConfigManager::get_changed_channels()
{
    std::lock_guard<std::mutex> lock(mutex_);
    return std::exchange(pending_changes_, {});
}

#define FETCH_REDIS_CONFIG_STRING_ITEM(VAR, FIELD_KEY, DEFAULT_VALUE)   \
    {                                                                   \
        if (const auto it = fields.find(FIELD_KEY); it != fields.end()) \
        {                                                               \
            VAR = it->second;                                           \
        }                                                               \
        else                                                            \
        {                                                               \
            VAR = DEFAULT_VALUE;                                        \
        }                                                               \
    }
#define FETCH_REDIS_CONFIG_INT_ITEM(VAR, FIELD_KEY, DEFAULT_VALUE)                                 \
    {                                                                                              \
        if (const auto it = fields.find(FIELD_KEY); it != fields.end())                            \
        {                                                                                          \
            try                                                                                    \
            {                                                                                      \
                VAR = std::stoll(it->second);                                                      \
            }                                                                                      \
            catch (...)                                                                            \
            {                                                                                      \
                std::cerr << "解析REDIS数值变量" #FIELD_KEY << "异常:" << it->second << std::endl; \
                VAR = DEFAULT_VALUE;                                                               \
            }                                                                                      \
        }                                                                                          \
        else                                                                                       \
        {                                                                                          \
            VAR = DEFAULT_VALUE;                                                                   \
        }                                                                                          \
    }
#define FETCH_REDIS_CONFIG_BOOL_ITEM(VAR, FIELD_KEY, DEFAULT_VALUE)     \
    {                                                                   \
        if (const auto it = fields.find(FIELD_KEY); it != fields.end()) \
        {                                                               \
            VAR = (to_lower(it->second) == "true");                     \
        }                                                               \
        else                                                            \
        {                                                               \
            VAR = DEFAULT_VALUE;                                        \
        }                                                               \
    }
#define FETCH_REDIS_CONFIG_LogLevel_ITEM(VAR, FIELD_KEY, DEFAULT_VALUE)                                    \
    {                                                                                                      \
        if (const auto it = fields.find(FIELD_KEY); it != fields.end())                                    \
        {                                                                                                  \
            try                                                                                            \
            {                                                                                              \
                VAR = LogLevel(parse_log_level(it->second));                                               \
            }                                                                                              \
            catch (...)                                                                                    \
            {                                                                                              \
                std::cerr << "解析REDIS枚举LogLevel变量" #FIELD_KEY << "异常:" << it->second << std::endl; \
                VAR = DEFAULT_VALUE;                                                                       \
            }                                                                                              \
        }                                                                                                  \
        else                                                                                               \
        {                                                                                                  \
            VAR = DEFAULT_VALUE;                                                                           \
        }                                                                                                  \
    }
OutputConfig ConfigManager::parse_config(const std::unordered_map<std::string, std::string> &fields) const
{
    OutputConfig config;

    FETCH_REDIS_CONFIG_BOOL_ITEM(config.console.enabled, "console.enabled", false);
    FETCH_REDIS_CONFIG_LogLevel_ITEM(config.console.min_level, "console.min_level", LogLevel::INFO_);

    FETCH_REDIS_CONFIG_BOOL_ITEM(config.file.enabled, "file.enabled", false);
    FETCH_REDIS_CONFIG_LogLevel_ITEM(config.file.min_level, "file.min_level", LogLevel::INFO_);
    FETCH_REDIS_CONFIG_STRING_ITEM(config.file.path_pattern, "file.file_path", "D:\\BridgeAnalysis_{:%Y%m%d}_{:04d}.log");
    FETCH_REDIS_CONFIG_INT_ITEM(config.file.max_size, "file.max_size", 100 * 1024 * 1024);
    FETCH_REDIS_CONFIG_INT_ITEM(config.file.max_files, "file.max_files", 10);
    FETCH_REDIS_CONFIG_INT_ITEM(config.file.rotation_interval, "file.rotation_interval", 86400);

    FETCH_REDIS_CONFIG_BOOL_ITEM(config.database.enabled, "database.enabled", false);
    FETCH_REDIS_CONFIG_LogLevel_ITEM(config.database.min_level, "database.min_level", LogLevel::INFO_);
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.type, "database.type", "mysql");
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.host, "database.host", "127.0.0.1");
    FETCH_REDIS_CONFIG_INT_ITEM(config.database.port, "database.port", 3306);
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.username, "database.username", "root");
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.password, "database.password", "123456");
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.database, "database.database", "BridgeAnalysis");
    FETCH_REDIS_CONFIG_STRING_ITEM(config.database.table, "database.table", "logs");

    return config;
}

// LogService实现
LogService::LogService(const std::string &redis_host, int32_t redis_port)
{
    sw::redis::ConnectionOptions opts;
    opts.host = redis_host;
    opts.port = redis_port;
    // 设置连接超时（TCP 连接建立的最长等待时间）
    opts.connect_timeout = std::chrono::milliseconds(2000); // 2秒
    // 设置操作超时（每次请求响应的最长等待时间）
    opts.socket_timeout = std::chrono::milliseconds(0); // 1秒

    // 其他优化配置
    opts.keep_alive = true; // 保持长连接
    opts.resp = 2;          // =3时使用 RESP3Redis 6+）

    redis_ = std::make_shared<sw::redis::Redis>(opts);
    config_manager_ = std::make_shared<ConfigManager>(*redis_);
    std::cout << "LogService: redis_host1=" << redis_host << ", redis_port1=" << redis_port << std::endl << std::flush;
    std::cout << "Host length: " << redis_host.length() << std::endl<< std::flush;
    try
    {
        auto reply = redis_->ping();
        std::cout << "Redis connection verified. PING reply: " << reply << std::endl;
    }
    catch (const sw::redis::TimeoutError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis TimeoutError error: " << e.what() << std::endl;
    }
    catch (const sw::redis::IoError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis IoError error: " << e.what() << std::endl;
    }
    catch (const std::exception &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "General error: " << e.what() << std::endl;
    }
}

LogService::~LogService()
{
    stop();
}

void LogService::start()
{
    running_ = true;
    sub_thread_ = std::thread(&LogService::subscription_thread, this);
    processing_thread_ = std::thread(&LogService::processing_thread, this);
}

void LogService::stop()
{
    running_ = false;
    if (sub_thread_.joinable())
        sub_thread_.join();
    if (processing_thread_.joinable())
        processing_thread_.join();

    std::lock_guard<std::mutex> lock(file_handles_mutex_);
    for (auto &[path, stream] : file_handles_)
    {
        if (stream.is_open())
        {
            stream.close();
        }
    }
}

void LogService::subscription_thread()
{
    try
    {
        auto sub = redis_->subscriber();
        auto onMessage = [this](auto channel, auto msg)
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            msg_queue_.emplace(std::string(channel), std::string(msg));
        };
        sub.on_message(onMessage);

        sub.subscribe("logging:channel:BridgeAnalysis");
        std::cout << "subscription_thread start..." << std::endl;
        while (running_)
        {
            try
            {
                sub.consume();
            }
            catch (const sw::redis::Error &e)
            {
                std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis subscription error: " << e.what() << std::endl;
                std::this_thread::sleep_for(1s);
            }
        }
    }
    catch (const sw::redis::TimeoutError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis TimeoutError error: " << e.what() << std::endl;
    }
    catch (const sw::redis::IoError &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Redis IoError error: " << e.what() << std::endl;
    }
    catch (const std::exception &e)
    {
        std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "General error: " << e.what() << std::endl;
    }
}

void LogService::processing_thread()
{
    std::cout << "processing_thread start..." << std::endl;
    while (running_)
    {
        apply_config_changes();

        std::queue<std::pair<std::string, std::string>> local_queue;
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            local_queue.swap(msg_queue_);
        }

        while (!local_queue.empty())
        {
            auto [channel, msg] = local_queue.front();
            local_queue.pop();

            LogLevel msg_level = LogLevel::INFO_;
            if (const size_t pos_start = msg.find("["), pos_end = msg.find("]"); pos_start != std::string::npos && pos_end != std::string::npos)
            {
                const std::string level_str = msg.substr(pos_start + 1, pos_end - pos_start - 1);
                msg_level = string_to_level(level_str);
            }

            std::vector<LogProcessor *> processors;
            {
                std::lock_guard<std::mutex> lock(processors_mutex_);
                if (const auto it = processors_.find(channel); it != processors_.end())
                {
                    for (auto &p : it->second)
                    {
                        processors.push_back(p.get());
                    }
                }
            }

            for (auto &processor : processors)
            {
                if (processor->should_process(msg_level))
                {
                    processor->process(msg, msg_level);
                }
            }
        }

        std::this_thread::sleep_for(100ms);
    }
}

void LogService::apply_config_changes()
{
    static auto throttleDuration = std::chrono::seconds(5);
    static std::chrono::time_point<std::chrono::steady_clock> lastPrintTime;
    auto now = std::chrono::steady_clock::now();
    if (now - lastPrintTime >= throttleDuration)
    {
        std::cout << "LogService::apply_config_changes." << std::endl;
        lastPrintTime = now;
        config_manager_->refresh();
    }

    const auto changed = config_manager_->get_changed_channels();

    for (const auto &channel : changed)
    {
        const auto config = config_manager_->get_config(channel);
        std::cout << "config:" << config << std::endl;
        std::vector<std::unique_ptr<LogProcessor>> new_processors;

        if (config.console.enabled)
        {
            new_processors.push_back(std::make_unique<ConsoleProcessor>(config));
        }

        if (config.file.enabled)
        {
            if (auto fp = create_file_processor(config.file, channel))
            {
                new_processors.push_back(std::move(fp));
            }
        }

        if (config.database.enabled)
        {
            if (auto dbp = create_db_processor(config.database, channel))
            {
                new_processors.push_back(std::move(dbp));
            }
        }

        std::lock_guard<std::mutex> lock(processors_mutex_);
        processors_[channel] = std::move(new_processors);
    }
}

static int32_t current_index_ = 1;

std::unique_ptr<LogProcessor> LogService::create_file_processor(const FileConfig &config, const std::string &channel)
{

    return std::make_unique<FileProcessor>(this, config, channel);
}

ConsoleProcessor::ConsoleProcessor(OutputConfig config) : config_(config) {};

void ConsoleProcessor::process(const std::string &msg, LogLevel)
{
    std::string msgGBK = boost::locale::conv::from_utf(msg.c_str(), std::string("gb2312"));
    std::cout << msgGBK << std::endl;
}

bool ConsoleProcessor::should_process(LogLevel msg_level) const
{
    return msg_level >= (LogLevel)config_.console.min_level;
}

std::unique_ptr<LogProcessor> ConsoleProcessor::clone() const
{
    return std::make_unique<ConsoleProcessor>(*this);
}

FileProcessor::FileProcessor(LogService *service, FileConfig cfg, std::string ch) : service_(service), config_(std::move(cfg)), channel_(std::move(ch)), last_rotation_(std::chrono::system_clock::now()) {}

FileProcessor::~FileProcessor()
{
    std::lock_guard<std::mutex> lock(service_->file_handles_mutex_);
    for (auto &[path, stream] : service_->file_handles_)
    {
        if (stream && stream.is_open())
        {
            stream.close();
        }
    }
}

void FileProcessor::process(const std::string &msg, LogLevel)
{
    int32_t add_size = (int32_t)msg.size() + 1;
    if (should_rotation())
    {
        rotate_files(true);
    }
    std::string path = generate_filename(last_rotation_, current_index_);
    std::lock_guard<std::mutex> lock(service_->file_handles_mutex_);
    {
        size_t current_size = 0;
        auto &stream = service_->file_handles_[path];
        if (stream.is_open())
        {
            current_size = stream.tellp();
            if (current_size + add_size >= config_.max_size)
            {
                rotate_files(false);
                path = generate_filename(last_rotation_, current_index_);
            }
        }
    }
    auto &stream = service_->file_handles_[path];
    if (!stream.is_open())
    {
        fs::create_directories(fs::path(path).parent_path());
        stream.open(path, std::ios::app);
    }
    if (stream.is_open())
    {
        stream << msg << '\n';
    }
}

bool FileProcessor::should_process(LogLevel msg_level) const
{
    return msg_level >= config_.min_level;
}

std::unique_ptr<LogProcessor> FileProcessor::clone() const
{
    // 使用拷贝构造函数创建新实例
    return std::make_unique<FileProcessor>(*this);
}

void FileProcessor::replace_placeholder(std::string &str, const std::string &placeholder, const std::string &fmt) const
{
    size_t pos = 0;
    while ((pos = str.find(placeholder, pos)) != std::string::npos)
    {
        str.replace(pos, placeholder.length(), fmt);
        pos += fmt.length();
    }
}
std::string FileProcessor::generate_filename(const std::chrono::system_clock::time_point &timepoint, int32_t index) const
{
    // 转换为本地时间点
    std::time_t tt = std::chrono::system_clock::to_time_t(timepoint);
    std::tm local_tm = localtime_thread_safe(tt);
    std::string pattern = config_.path_pattern;
    replace_placeholder(pattern, "%%date%%",
                        (boost::format("%04d%02d%02d") % (local_tm.tm_year + 1900) % (local_tm.tm_mon + 1) % local_tm.tm_mday).str());
    replace_placeholder(pattern, "%%index%%",
                        (boost::format("%04d") % index).str());
    return pattern;
}

bool FileProcessor::should_rotation() const
{
    const auto now = std::chrono::system_clock::now();
    const auto duration = std::chrono::duration_cast<std::chrono::seconds>(now - last_rotation_).count();
    if (duration >= (int32_t)config_.rotation_interval || generate_filename(last_rotation_, 1) != generate_filename(now, 1))
    {
        return true;
    }
    return false;
}

void FileProcessor::rotate_files(bool byTime)
{
    // 关闭当前文件
    const std::string current_path = generate_filename(last_rotation_, current_index_);
    if (auto it = service_->file_handles_.find(current_path); it != service_->file_handles_.end())
    {
        auto stream = std::move(it->second);
        stream.close();
        auto wtime = boost::filesystem::last_write_time(current_path);
        service_->file_handles_his_.emplace(current_path, wtime);
        service_->file_handles_.erase(it);
    }

    // 清理旧文件
    std::vector<std::pair<std::string, std::time_t>> files;
    for (const auto &[path, wtime] : service_->file_handles_his_)
    {
        if (path.find(channel_.substr(16)) != std::string::npos)
        {
            files.push_back(std::pair<std::string, std::time_t>(path, wtime));
        }
    }
    std::sort(files.begin(), files.end(), [](const auto &a, const auto &b)
              { return a.second < b.second; });
    while (files.size() >= config_.max_files)
    {
        try
        {
            fs::remove(files.front().first);
        }
        catch (const std::exception &e)
        {
            std::cerr << _S_("[%s/%s-%d]", __FILE__, __FUNCTION__, __LINE__) << "Fatal error while removing log files：" << e.what() << std::endl;
        }
        files.erase(files.begin());
    }

    if (byTime)
    {
        last_rotation_ = std::chrono::system_clock::now();
        current_index_ = 1;
    }
    else
    {
        current_index_++;
    }
}

MySQLProcessor::MySQLProcessor(LogService *service, DatabaseConfig cfg, std::string ch)
    : service_(service), config_(std::move(cfg)), channel_(std::move(ch))
{
    // 初始化连接池
    for (int i = 0; i < 3; ++i)
    {
        auto conn = createConnection();
        if (conn && conn->isConnected())
        {
            connections_pool_.push_back(std::move(conn));
        }
    }
}
MySQLProcessor::MySQLProcessor(const MySQLProcessor &other)
    : service_(other.service_),
      config_(other.config_),
      channel_(other.channel_)
{
    // 复制连接池（创建新的连接）
    std::lock_guard<std::mutex> lock(other.connections_mutex_);
    for (const auto &conn : other.connections_pool_)
    {
        if (conn && conn->isConnected())
        {
            auto new_conn = createConnection();
            if (new_conn && new_conn->isConnected())
            {
                connections_pool_.push_back(std::move(new_conn));
            }
        }
    }
}
MySQLProcessor::~MySQLProcessor()
{
    std::lock_guard<std::mutex> lock(connections_mutex_);
    connections_pool_.clear();
}

std::unique_ptr<DBWrapper> MySQLProcessor::createConnection()
{
    DBWrapper::Config config{
        config_.host,
        config_.port,
        config_.username,
        config_.password,
        config_.database};

    auto wrapper = std::make_unique<DBWrapper>(config);
    if (!wrapper->connect())
    {
        return nullptr;
    }
    return wrapper;
}

void MySQLProcessor::process(const std::string &msg, LogLevel level)
{
    std::unique_ptr<DBWrapper> conn;
    {
        std::lock_guard<std::mutex> lock(connections_mutex_);
        if (!connections_pool_.empty())
        {
            conn = std::move(connections_pool_.back());
            connections_pool_.pop_back();
        }
    }

    if (!conn)
    {
        conn = createConnection();
    }

    if (!conn || !conn->isConnected())
    {
        std::cerr << "Failed to get valid database connection" << std::endl;
        return;
    }

    std::string query = "INSERT INTO " + config_.table +
                        " (channel, level, message, timestamp) VALUES (?, ?, ?, NOW())";

    std::vector<std::string> params = {
        channel_,
        log_level_string(level),
        msg};

    if (!conn->prepareAndExecute(query, params))
    {
        std::cerr << "Failed to execute prepared statement" << std::endl;
    }

    {
        std::lock_guard<std::mutex> lock(connections_mutex_);
        connections_pool_.push_back(std::move(conn));
    }
}

bool MySQLProcessor::should_process(LogLevel msg_level) const
{
    return msg_level >= config_.min_level;
}

std::unique_ptr<LogProcessor> MySQLProcessor::clone() const
{
    return std::unique_ptr<LogProcessor>(new MySQLProcessor(*this));
}

std::unique_ptr<LogProcessor> LogService::create_db_processor(const DatabaseConfig &config, const std::string &channel)
{
    try
    {
        auto processor = std::make_unique<MySQLProcessor>(this, config, channel);

        // 测试连接并创建表结构
        auto conn = processor->createConnection();
        if (!conn || !conn->isConnected())
        {
            throw std::runtime_error("Failed to connect to database");
        }

        std::string createTable =
            "CREATE TABLE IF NOT EXISTS " + config.table + "("
                                                           "  id INT AUTO_INCREMENT PRIMARY KEY,"
                                                           "  channel VARCHAR(255) NOT NULL,"
                                                           "  level VARCHAR(10) NOT NULL,"
                                                           "  message TEXT NOT NULL,"
                                                           "  timestamp DATETIME NOT NULL,"
                                                           "  INDEX idx_channel (channel),"
                                                           "  INDEX idx_timestamp (timestamp)"
                                                           ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4";

        if (!conn->execute(createTable))
        {
            throw std::runtime_error("Failed to create log table");
        }

        // 将连接放回池中
        {
            std::lock_guard<std::mutex> lock(processor->get_conn_mutex());
            processor->get_conn_pool().push_back(std::move(conn));
        }

        return processor;
    }
    catch (const std::exception &e)
    {
        std::cerr << "Failed to initialize database processor for channel ["
                  << channel << "]: " << e.what() << std::endl;
        return nullptr;
    }
}

int32_t main(int32_t argc, char **argv)
{
    for (int i = 0; i < argc; ++i)
    {
        if (std::string(argv[i]) == "--version")
        {
            std::cout << "LogService version: " << LOGSERVICE_VERSION << std::endl;
            return 0;
        }
    }
    ConfigINI config;
    std::string redisHost = config.GetString("RedisHost", "127.0.0.1");
    int32_t redisPort = config.GetInt("RedisPort", 6379);
    LogService service(redisHost, redisPort);
    service.start();

    // 保持主线程运行
    while (true)
        std::this_thread::sleep_for(1s);

    service.stop();
    return 0;
}